package cn.taqu.es.util;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.ActionRequestBuilder;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.search.SearchHit;
import org.nlpcn.es4sql.query.DefaultQueryAction;
import org.nlpcn.es4sql.query.ESActionFactory;
import org.nlpcn.es4sql.query.SqlElasticSearchRequestBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;

/**
 * Rest high level client
 */
@Component
@Slf4j
public class ElasticSearchUtil {

    @Autowired
    private RestHighLevelClient rhlClient;

    /**
     * 按sql查询
     *
     * @param startPage
     * @param pageSize
     * @param sql
     * @return
     * @throws Exception
     */
    public EsPage searchBySql(Integer startPage, Integer pageSize, String sql) throws Exception {
        String limitSql = String.format("%s limit %d, %d", sql, startPage * pageSize, pageSize);
        DefaultQueryAction queryAction = (DefaultQueryAction) ESActionFactory.create(null, limitSql);
        SqlElasticSearchRequestBuilder builder = queryAction.explain();
        ActionRequestBuilder actionRequestBuilder = builder.getBuilder();
        ActionRequest actionRequest = actionRequestBuilder.request();
        SearchRequest searchRequest = (SearchRequest) actionRequest;
        SearchResponse searchResponse = rhlClient.search(searchRequest, RequestOptions.DEFAULT);
        return parsePageResult(startPage, pageSize, searchResponse);
    }


    /**
     * 新增操作
     *
     * @param index
     * @param type
     * @param id
     * @param json
     * @param appcode
     * @return
     * @throws IOException
     */
    public String create(String index, String type, String id, String json, Integer appcode) throws IOException {
        IndexRequest indexRequest = new IndexRequest(index, type, id);
        Map jsonMap = JSONObject.parseObject(json, new TypeReference<Map>() {
        }.getType());
        indexRequest.source(jsonMap, XContentType.JSON);
        indexRequest.versionType(VersionType.EXTERNAL);
        indexRequest.version(new Date().getTime());
        IndexResponse response = rhlClient.index(indexRequest, RequestOptions.DEFAULT);
        log.info("插入成功-index {{}} type {{}} id {{}}", index, type, id);
        return response.getId();

    }

    /**
     * 更新操作
     *
     * @param index
     * @param type
     * @param id
     * @param json
     * @return
     * @throws IOException
     */
    public String update(String index, String type, String id, String json) throws IOException {
        UpdateRequest request = new UpdateRequest(index, type, id);
        Map jsonMap = JSONObject.parseObject(json, new TypeReference<Map>() {
        }.getType());
        request.doc(jsonMap);
        UpdateResponse response = rhlClient.update(request, RequestOptions.DEFAULT);
        log.info("更新索引成功-index{},id{},json{}", index, id, json);
        return response.getId();
    }


    public void batchData(String index, String type, Map<String, Object> map) throws IOException {

        BulkRequest request = new BulkRequest(index, type);
        map.keySet().forEach(key -> {
            IndexRequest indexRequest = new IndexRequest();
            indexRequest.id(key);
            Map jsonMap = JSONObject.parseObject(map.get(key).toString(), Map.class);
            indexRequest.source(jsonMap, XContentType.JSON);
            indexRequest.versionType(VersionType.EXTERNAL);
            indexRequest.version(new Date().getTime());
            request.add(indexRequest);
        });
        rhlClient.bulk(request, RequestOptions.DEFAULT);
        log.info("批量插入成功-index {{}} ", index);

    }

    /**
     * @param searchResponse 设置返回值
     */
    private static List<String> setSearchResponse(SearchResponse searchResponse) {
        List<String> sourceList = new ArrayList<>();
        for (SearchHit searchHit : searchResponse.getHits().getHits()) {
            sourceList.add(searchHit.getId());
            //这里可以查询出需要返回的字段进行填充,这里只拿id，然后拿到id再去库里查询最新的数据
        }
        return sourceList;
    }

    /**
     * 封装返回对象
     *
     * @param startPage
     * @param pageSize
     * @param searchResponse
     * @return
     */
    private static EsPage parsePageResult(Integer startPage, Integer pageSize, SearchResponse searchResponse) {
        if (searchResponse.status().getStatus() == 200) {
            long totalHits = searchResponse.getHits().totalHits;
            long length = searchResponse.getHits().getHits().length;
            log.info("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length);

            // 解析对象
            List<String> sourceList = setSearchResponse(searchResponse);
            return new EsPage(startPage, pageSize, (int) totalHits, sourceList);
        }

        log.error("es搜索数据失败, http响应状态码:{}错误", searchResponse.status().getStatus());
        return new EsPage(startPage, pageSize, 0, Collections.EMPTY_LIST);
    }


}
